package drds.global_transaction.api.rpc.rpc_server;

import drds.common.ShouldNeverHappenException;
import drds.global_transaction.api.api.RpcContext;
import drds.global_transaction.api.in_bound_handler.TransactionMessageHandler;
import drds.global_transaction.api.message_codec.*;
import drds.global_transaction.api.message_codec.request_and_response.register_manager.resource_manager.RegisterResourceManagerRequestMessage;
import drds.global_transaction.api.message_codec.request_and_response.register_manager.resource_manager.RegisterResourceManagerResponseMessage;
import drds.global_transaction.api.message_codec.request_and_response.register_manager.transaction_manager.RegisterTransactionManagerRequestMessage;
import drds.global_transaction.api.message_codec.request_and_response.register_manager.transaction_manager.RegisterTransactionManagerResponseMessage;
import drds.global_transaction.common.thread.NamedThreadFactory;
import drds.global_transaction.common.util.Nets;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.*;

@Slf4j
public class TransactionMessageListenerImpl implements TransactionMessageListener
{
	class BatchSaveLogRunnable implements Runnable
	{

		@Override
		public void run()
		{
			while (true)
			{
				if (messageQueue.size() > 0)
				{
					StringBuilder sb = new StringBuilder();
					while (!messageQueue.isEmpty())
					{
						sb.append(messageQueue.poll()).append(batch_log_split);
					}
					if (log.isInfoEnabled())
					{
						log.info(sb.toString());
					}
				}
				try
				{
					Thread.sleep(idle_check_mills);
				} catch (InterruptedException exx)
				{
					log.error(exx.getMessage());
				}
			}
		}
	}

	private static final int max_log_send_thread = 1;
	private static final long keep_alive_time = 0l;
	private static final String thread_prefix = "batch_logger_print";
	private static final long idle_check_mills = 3l;
	private static final String batch_log_split = "\n";
	private static BlockingQueue<String> messageQueue = new LinkedBlockingQueue<String>();
	@Setter
	@Getter
	private TransactionMessageHandler transactionMessageHandler;

	@Setter
	@Getter
	private MessageSend messageSend;

	public TransactionMessageListenerImpl(TransactionMessageHandler transactionMessageHandler)
	{
		this.transactionMessageHandler = transactionMessageHandler;
	}

	public void init()
	{
		ExecutorService executorService = new ThreadPoolExecutor(max_log_send_thread, //
				max_log_send_thread, //
				keep_alive_time, //
				TimeUnit.MILLISECONDS, //
				new LinkedBlockingQueue<Runnable>(), //
				new NamedThreadFactory(thread_prefix, max_log_send_thread, true));//
		executorService.submit(new BatchSaveLogRunnable());//
	}

	@Override
	public void onPing(ChannelHandlerContext channelHandlerContext, MessageSend messageSend, long messageId)
	{
		try
		{
			messageSend.synchronousSendResponse(channelHandlerContext.channel(), messageId, HeartbeatMessage.pong);
		} catch (Throwable throwable)
		{
			log.error("", "send response error", throwable);
		}
		if (log.isDebugEnabled())
		{
			log.debug("received ping from " + channelHandlerContext.channel().remoteAddress());
		}
	}

	@Override
	public void onRegisterResourceManagerRequest(ChannelHandlerContext channelHandlerContext, MessageSend messageSend, long messageId, RegisterResourceManagerRequestMessage registerResourceManagerRequest, RegisterCheckAuthHandler registerCheckAuthHandler)
	{

		boolean isSuccess = false;
		try
		{
			if (null == registerCheckAuthHandler || null != registerCheckAuthHandler && registerCheckAuthHandler.registerResourceManagerCheckAuth(registerResourceManagerRequest))
			{
				ChannelManager.registerResourceManagerClientChannel(registerResourceManagerRequest, channelHandlerContext.channel());
				Version.putChannelVersion(channelHandlerContext.channel(), registerResourceManagerRequest.getVersion());
				isSuccess = true;
			}
		} catch (Exception exx)
		{
			isSuccess = false;
			log.error(exx.getMessage());
		}
		messageSend.synchronousSendResponse(channelHandlerContext.channel(), messageId, new RegisterResourceManagerResponseMessage(isSuccess));
		if (log.isInfoEnabled())
		{
			log.info("resource_manager_client register success,message:" + registerResourceManagerRequest + ",channel" + ":" + channelHandlerContext.channel());
		}
	}

	@Override
	public void onRegisterTransactionManagerRequest(ChannelHandlerContext channelHandlerContext, MessageSend messageSend, long messageId, RegisterTransactionManagerRequestMessage registerTransactionManagerRequest, RegisterCheckAuthHandler registerCheckAuthHandler)
	{
		String ipAndPort = Nets.getHostAddressAndPort(channelHandlerContext.channel().remoteAddress());
		Version.putChannelVersion(channelHandlerContext.channel(), registerTransactionManagerRequest.getVersion());
		boolean isSuccess = false;
		try
		{
			if (null == registerCheckAuthHandler || null != registerCheckAuthHandler && registerCheckAuthHandler.//
					registerTransactionManagerCheckAuth(registerTransactionManagerRequest))
			{
				ChannelManager.registerTransactionManagerClientChannel(registerTransactionManagerRequest, channelHandlerContext.channel());
				Version.putChannelVersion(channelHandlerContext.channel(), registerTransactionManagerRequest.getVersion());
				isSuccess = true;
				if (log.isInfoEnabled())
				{
					log.info(String.format("checkAuth for client:%s vgroup:%s ok", ipAndPort, registerTransactionManagerRequest.getTransactionServiceGroup()));
				}
			}
		} catch (Exception exx)
		{
			isSuccess = false;
			log.error(exx.getMessage());
		}
		messageSend.synchronousSendResponse(channelHandlerContext.channel(), messageId, new RegisterTransactionManagerResponseMessage(isSuccess));
	}

	@Override
	public void onMessage(ChannelHandlerContext channelHandlerContext, MessageSend messageSend, long messageId, Object message)
	{
		//
		//Channel channel=ChannelManager.channelThreadLocal.get();
		//RpcContext rpcContext = ChannelManager.getRpcContext(channel);
		RpcContext rpcContext = ChannelManager.getRpcContext(channelHandlerContext.channel());

		if (log.isDebugEnabled())
		{
			log.debug("transaction_coordinator received:" + message + ",clientIp:" + Nets.getHostAddress(channelHandlerContext.channel().remoteAddress()) + ",vgroup:" + rpcContext.getTransactionServiceGroup());
		} else
		{
			messageQueue.offer(message + ",clientIp:" + Nets.getHostAddress(channelHandlerContext.channel().remoteAddress()) + ",vgroup:" + rpcContext.getTransactionServiceGroup());
		}
		if (!(message instanceof Message))
		{
			return;
		}
		if (message instanceof RequestMessages)
		{
			ResponseMessage[] responseMessages = new ResponseMessage[((RequestMessages) message).requestMessageList.size()];
			for (int i = 0; i < responseMessages.length; i++)
			{
				final RequestMessage requestMessage = ((RequestMessages) message).requestMessageList.get(i);
				responseMessages[i] = transactionMessageHandler.onRequest(rpcContext, requestMessage);
			}
			ResponseMessages $responseMessages = new ResponseMessages();
			$responseMessages.setResponseMessages(responseMessages);
			messageSend.synchronousSendResponse(channelHandlerContext.channel(), messageId, $responseMessages);
		} else if (message instanceof ResponseMessage)
		{
			transactionMessageHandler.onResponse(rpcContext, (ResponseMessage) message);
		}else {
			throw new ShouldNeverHappenException();
		}
	}

}
